package com.atguigu.dw.gmall.realtime.app

import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.JSON
import com.atguigu.constan.ConstanVal
import com.atguigu.dw.gmall.realtime.bean.StartupLog
import com.atguigu.dw.gmall.realtime.util.{MyKafkaUtil, RedisUtil}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * description ：统计日活app
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/11 18:43
 * modified By ：
 * version:    : 1.0
 */
object DauApp {

  private var sparkContext: SparkContext = null

  def main(args: Array[String]): Unit = {

    // 创建 StreamContext
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName(DauApp.getClass.getName)
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    sparkContext = ssc.sparkContext

    // 读取 kafkaStream 数据
    val dawStream = MyKafkaUtil.getKafkaStream(ssc, Set[String](ConstanVal.KAFKA_TOPIC_START))

    // 将 kafkaStream 数据转换为样例类
    val dawBean: DStream[StartupLog] = dawStream.map({
      case (_, v) => JSON.parseObject(v, classOf[StartupLog])
    })

    // 对数据时行去重和入库 hbase
    dawBean.foreachRDD(rdd => {
      rdd.foreachPartition(it => {
        // 1、将数据使用 redis 去重
        val client = RedisUtil.getJedisClient
        it.foreach(log => {
          if (client.sadd(ConstanVal.KAFKA_TOPIC_START + ":" + new SimpleDateFormat("yyyy-MM-dd").format(new Date()), log.mid) == 1) {
            // 2、入库hbase
            import org.apache.phoenix.spark._
            sparkContext.makeRDD(List(log)).saveToPhoenix(
              ConstanVal.GMALL_HBASE_TABLE_NAME,
              Seq("MID", "UID", "APPID", "AREA", "OS", "CHANNEL", "LOGTYPE", "VERSION", "TS", "LOGDATE", "LOGHOUR"),
              zkUrl = Some(ConstanVal.PHOENIX_ZK_ADRESS)
            )
          }
        })
        client.close()
      })
    })

    // 启动ssc,将阻止程序停止
    ssc.start()
    ssc.awaitTermination()
  }
}
